package com.huan.maket;


import com.huan.bean.MarketingUserBehavior;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Arrays;
import java.util.List;
import java.util.Random;


//实现自定义的模拟市场用户行为数据源
public class SimulatedMarketingBehaviorSource implements SourceFunction<MarketingUserBehavior> {
    // 控制是否运行的标识位
    boolean isRunning = true;

    //定义行为和渠道的范围
    //TODO 点击,下载,安装,卸载
    String[] behaviors = {"CLICK","DOWNLOAD","INSTALL","UNINSTALL"};
    String[] channels = {"app store","wechat","weibo"};
    List<String> behaviorList =  Arrays.asList(behaviors);
    List<String> channelList = Arrays.asList(channels);

    //随机数
    Random random = new Random();
    
    @Override
    public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {
        while (isRunning) {
            //随机生成所有字段
            long id = random.nextLong();
            /**
             * get()返回此列表中指定位置上的元素。
             * random.返回当前随机数
             * 获取behaviorList对象中指定位置上(random 当前随机数，随机数指(behaviorList对象中的某一个))的元素
             */
            String behavior = behaviorList.get( random.nextInt( behaviorList.size() ) );
            String channel = channelList.get( random.nextInt( channelList.size() ) );
            long timestamp = System.currentTimeMillis();

            //输出
            ctx.collect( new MarketingUserBehavior( id, behavior, channel, timestamp ) );
        }

        Thread.sleep( 1000L );
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
